// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jaewon

#include "coin1/fastfeed/proto_feed_builder.h"

#include <glog/logging.h>

#include "coin1/base/exception_util.h"
#include "coin1/fastfeed/view/json_object_table_builder.h"
#include "coin1/fastfeed/view/string_json_builder.h"

namespace fastfeed {
namespace {

struct CallbackController {
  void operator()(const view::ViewBase* view) const {
    if (*call) {
      callback(view);
    }
  }
  std::function<void(const view::ViewBase*)> callback;
  bool* call;
};  // struct CallbackController

void RegisterDefaultViews() {
  view::RegisterViewType<view::BookBuilder>("book", true);
  view::RegisterViewType<view::TradeBuilder>("trade", true);
  view::RegisterViewType<view::StringJsonBuilder>("string", true);
  view::RegisterViewType<view::StringJsonBuilder>("json", true);
  view::RegisterViewType<view::JsonObjectTableBuilder>("json_object_table", true);
}

}  // namespace

FeedBuilderBase::FeedBuilderBase(const std::string& sub_req) : sub_req_(sub_req) {}

void FeedBuilderBase::SubscribeBook(
    const std::string& symbol,
    std::function<void(const view::BookBuilder*)> callback) {
  Subscribe<view::BookBuilder>(proto::FEED_TYPE_BOOK, "", symbol, "book", callback);
}

void FeedBuilderBase::SubscribeTrade(
    const std::string& symbol,
    std::function<void(const view::TradeBuilder*)> callback) {
  Subscribe<view::TradeBuilder>(proto::FEED_TYPE_TRADE, "", symbol, "trade", callback);
}

ProtoFeedBuilder::ProtoFeedBuilder() { RegisterDefaultViews(); }

ProtoFeedBuilder::ProtoFeedBuilder(const std::string& sub_req) : FeedBuilderBase(sub_req) {
  RegisterDefaultViews();
}

std::string ProtoFeedBuilder::ResolveViewType(
    proto::FeedType feed_type,
    const std::string& custom_feed_type,
    const std::string& symbol) {
  switch (feed_type) {
    case proto::FEED_TYPE_BOOK:
      return "book";

    case proto::FEED_TYPE_TRADE:
      return "trade";

    default:
      LOG(ERROR) << "Unable to resolve view type.";
      return std::string();
  }
}

void ProtoFeedBuilder::Subscribe(
    proto::FeedType feed_type,
    const std::string& custom_feed_type,
    const std::string& symbol,
    const std::string& view_type,
    std::function<void(const view::ViewBase*)> callback) {
  if (feed_type != proto::FEED_TYPE_CUSTOM) {
    CHECK_THROW(custom_feed_type.empty(), std::invalid_argument) << custom_feed_type;
  } else {
    CHECK_THROW(!custom_feed_type.empty(), std::invalid_argument);
  }

  std::string final_view_type = view_type;
  if (final_view_type.empty()) {
    final_view_type = ResolveViewType(feed_type, custom_feed_type, symbol);
  }
  CHECK_THROW(!final_view_type.empty(), std::logic_error);

  view::ViewKey view_key {feed_type, custom_feed_type, symbol, final_view_type};
  view_collection_.SubscribeView(
      view_key,
      CallbackController {callback, &invoke_user_callbacks_flag_});
}

void ProtoFeedBuilder::UpdateBySerializedProto(std::string_view serialized) {
  bool res = feed_cache_.ParseFromArray(serialized.data(), serialized.size());
  if (!res) {
    LOG(ERROR) << "Protobuf parse failed. Skipping.";
    return;
  }
  return UpdateByProto(feed_cache_);
}

void ProtoFeedBuilder::UpdateByProto(const proto::Feed& feed) {
  if (!invoke_user_callbacks_ ||
      (from_timestamp_.has_value() && feed.timestamp() < from_timestamp_.value())) {
    invoke_user_callbacks_flag_ = false;
  } else {
    invoke_user_callbacks_flag_ = true;
  }
  view_collection_.InvokeFeedCallbacks(feed);
}

}  // namespace fastfeed
